package com.atguigu.flink.state.keyedstate;

import com.atguigu.flink.func.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by 黄凯 on 2023/6/20 0020 21:06
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 计算每种传感器的水位和
 */
public class Flink09_ReducingState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
                .socketTextStream("127.0.0.1", 8888)
                .map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedDS = ds.keyBy(WaterSensor::getId);

        keyedDS
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    private ReducingState<Integer> sumVc;

                    //初始化方法
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //找到状态的备份存储,取出状态并赋值
                        sumVc = getRuntimeContext().getReducingState(new ReducingStateDescriptor<>("sumVc",
                                new ReduceFunction<Integer>() {
                                    // reduce: 两两聚合。 输入和输出类型是一致的
                                    @Override
                                    public Integer reduce(Integer value1, Integer value2) throws Exception {
                                        return value1 + value2;
                                    }
                                },
                                Integer.class));
                    }


                    @Override
                    public void processElement(WaterSensor waterSensor,
                                               KeyedProcessFunction<String, WaterSensor, String>.Context context,
                                               Collector<String> collector) throws Exception {

                        //只要存进去，自动聚合
                        sumVc.add(waterSensor.getVc());

                        collector.collect(context.getCurrentKey() + " 累积的vc和:" + sumVc.get());

                    }
                }).print();

        env.execute();

    }

}
